package com.tiantang.kafka.stream;

import java.util.Properties;

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import org.apache.kafka.streams.processor.TopologyBuilder;

public class MyKafkaStreams {
	
	/**
	 * 
	 * @Title: process  
	 * @description:功能描述 流式处理消息
	 * @param sourceTopic 源Topic
	 * @param toTopic   处理完消息之后消息发送至目标Topicß
	 * @author:liujinkun
	 * @date 2018年5月28日
	 */
	public void process(String sourceTopic,String toTopic){
		// 1. kafka stream 配置
		Properties properties =new Properties();
		properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "logProcessor");
		properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "com.tiantang.hadoop01:9092");
		
		// 2. 创建StreamConfig
		StreamsConfig streamsConfig=new StreamsConfig(properties);
		
		// 3. 创建拓扑结构
		TopologyBuilder builder=new TopologyBuilder();
		builder.addSource("SOURCE", sourceTopic).addProcessor("LOGPROCESSOR", new ProcessorSupplier<byte[], byte[]>() {

			public Processor<byte[], byte[]> get() {
				return new LogProcessor();
			}
		}, "SOURCE").addSink("SINK", toTopic, "LOGPROCESSOR");
		
		// 4. 实例化 KafkaStreams对象
		KafkaStreams stream=new KafkaStreams(builder, streamsConfig);
		// 开始流式处理消息
		stream.start();
	}
	
	// JUNIT 不支持多线程的测试，因此在此写main方法进行测试
	public static void main(String[] args) {
		MyKafkaStreams streams=new MyKafkaStreams();
		streams.process("test", "first");
	}

}
